package com.zhang.flink.state;

import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @title: 广播状态case
 * @author: zhang
 * @date: 2022/2/21 18:47
 */
public class BroadStateCase {
    public static void main(String[] args) throws Exception {
        //todo 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);

        DataStreamSource<String> source = env
                .socketTextStream("localhost", 9999);

        DataStreamSource<String> broadStream = env
                .socketTextStream("localhost", 8888);

        //定义状态并广播
        MapStateDescriptor<String, String> mapStateDescriptor = new MapStateDescriptor<>("map-state", Types.STRING, Types.STRING);
        // 广播流
        BroadcastStream<String> broadcast = broadStream.broadcast(mapStateDescriptor);

        source
                .connect(broadcast)
                .process(new BroadcastProcessFunction<String, String, String>() {
                    @Override
                    public void processElement(String value, BroadcastProcessFunction<String, String, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {
                        String aSwitch = ctx.getBroadcastState(mapStateDescriptor).get("Switch");
                        if ("1".equals(aSwitch)){
                            out.collect("切换到----1");
                        }else if ("2".equals(aSwitch)){
                            out.collect("切换到----2");
                        }else {
                            out.collect("切换到----其他");
                        }
                    }

                    //todo 广播流处理逻辑
                    @Override
                    public void processBroadcastElement(String value, BroadcastProcessFunction<String, String, String>.Context ctx, Collector<String> out) throws Exception {
                        ctx.getBroadcastState(mapStateDescriptor).put("Switch",value);
                    }
                })
                .print();

        //todo 执行任务
        env.execute();

    }
}
